package haoyoutj.com;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.File;
import java.io.IOException;


public class Friends {
    public static class FriendsMap extends Mapper<LongWritable, Text,Text,Text>{
        Text k = new Text();
        Text v = new Text();
     @Override
        protected void map(LongWritable key,Text value,Context context)throws IOException,InterruptedException{
// 空格切分，有时编码格式问题导致错误，要写成\t
         String[] s = value.toString().split("\t");
         /**
          * 这里取到的值为B-B A
          * B-C A
          * B-D A
          * C-B A
          * C-C A
          * C-D A
          * A-A C
          * A-B C
          *以空格切分的话得到的是B-C和A，与前面的-无关
          */
          k.set(s[0]);
          v.set(s[1]);
          context.write(k,v);
        }
    }
        public static class FriendsReduce extends Reducer<Text,Text,Text,Text>{
        @Override
            protected void reduce (Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException{

            StringBuffer buffer = new StringBuffer();
            for (Text value :values){
                buffer.append(value+",");//拼接作用，以逗号隔开
            }
            context.write(key,new Text(buffer.toString()));
        }
        }
    /**
     * 结果是：
     *A-A C, 由于所用例子数据少，所以逗号后面没有东西
     *A-B C,
     *B-B A,
     *B-C A,
     *B-D A,
     *C-B A,
     *C-C A,
     *C-D A,
     */
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        try {
            Job job = Job.getInstance(conf);

            job.setJarByClass(Friends.class);

            job.setMapperClass(FriendsMap.class);
            job.setReducerClass(FriendsReduce.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            //查看文件是否存在
            File file = new File("F:\\haoyou");
            if (file.exists()){
                FileUtil.fullyDelete(file);
            }
            FileInputFormat.addInputPath(job,new Path("F:/file/output"));
            FileOutputFormat.setOutputPath(job,new Path("F:\\haoyou"));
            job.setNumReduceTasks(1);

            System.exit(job.waitForCompletion(true)?0 : 1);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
